package com.whz.bus.debezium;

import com.google.common.collect.Lists;
import com.whz.generic.threads.DefaultCommonThreadPool;
import io.debezium.engine.DebeziumEngine;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.SmartLifecycle;
import org.springframework.lang.NonNull;
import org.springframework.util.Assert;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

@Slf4j
public class DebeziumServerBootstrap<R> implements InitializingBean, SmartLifecycle {

    private final ExecutorService executor =
            DefaultCommonThreadPool.DEFAULT_THREAD_POOL.defaultAsyncExecutorService("bus-cdc", 1, 2, 30000, 200);

    private final List<DebeziumEngine<R>> debeziumEngine = Lists.newLinkedList();

    public DebeziumServerBootstrap(@NonNull List<DebeziumEngine<R>> debeziumEngine) {
        this.debeziumEngine.addAll(debeziumEngine);
    }

    @Override
    public void start() {
        debeziumEngine.forEach(executor::execute);
    }

    @Override
    public void stop() {
        try {
            executor.shutdown();
            while (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
                log.info("Waiting another 5 seconds for the embedded engine to shut down");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    @Override
    public boolean isRunning() {
        return false;
    }

    @Override
    public void afterPropertiesSet() {
        Assert.notEmpty(debeziumEngine, "debeziumEngine must not be null");
    }
}